package com.liu.queue.storage.impl;

import com.liu.common.Constants;
import com.liu.queue.message.MessageTypeEnum;
import com.liu.queue.message.PushMessage;
import com.liu.queue.message.StoragePushMessage;
import com.liu.queue.storage.Storage;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import org.bson.Document;

import java.io.IOException;
import java.io.InputStream;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 普通消息
 *
 * @param
 * @BelongsProject: QueueDemo2
 * @BelongsPackage: com.liu.queue.cache
 * @author: L
 * @Date: 2023/10/17 10:15
 * @return
 */
public class OrdinaryStorage implements Storage {
    // topickey,topic对应的队列为value--->每一个主题对应一个队列
    // 频繁在队列尾插入数据,队列头删除数据,ArrayBlockingQqueue性能会更好一些
    private static volatile ConcurrentHashMap<Integer, ArrayBlockingQueue<StoragePushMessage>> topicToMessage;
    // 容量
    private volatile Integer capacity;

    private OrdinaryStorage() {
        Properties properties = new Properties();
        InputStream inStream = Thread.currentThread().getContextClassLoader().getResourceAsStream("application.properties");
        try {
            properties.load(inStream);
        } catch (IOException e) {
            throw new RuntimeException("没有application.properties文件");
        }
        String property = properties.getProperty("cache.capacity", "1000");
        capacity = Integer.valueOf(property);
        topicToMessage = new ConcurrentHashMap<>();
    }

    @Override
    public void cache(PushMessage pushMessage) {
        Integer key = new String(pushMessage.getTopic()).hashCode();
        ArrayBlockingQueue<StoragePushMessage> orDefault = topicToMessage.getOrDefault(key, new ArrayBlockingQueue<>(capacity));
        // 抛异常
        StoragePushMessage storagePushMessage = new StoragePushMessage();
        storagePushMessage.setMessageID(pushMessage.getMessageID());
        storagePushMessage.setSerializationType(pushMessage.getSerializationType().getType());
        storagePushMessage.setMessageType(pushMessage.getMessageType().getType());
        String t = null;
        String b = null;
        try {
            byte[] topic = pushMessage.getTopic();
            t = new String(topic, "UTF-8");
            byte[] body = pushMessage.getBody();
            b = new String(body,"UTF-8");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        storagePushMessage.setTopic(t);
        storagePushMessage.setBody(b);
        orDefault.add(storagePushMessage);
        topicToMessage.put(key, orDefault);
    }

    @Override
    public boolean support(MessageTypeEnum messageType) {
        if (messageType==MessageTypeEnum.PUSH_ORDINARY_REQUEST || messageType==MessageTypeEnum.PULL_ORDINARY_REQUEST){
            return true;
        }
        return false;
    }

    @Override
    public List<StoragePushMessage> get(byte[] topic, Integer total) {
        ArrayBlockingQueue<StoragePushMessage> queue = topicToMessage.get(new String(topic).hashCode());
        if (queue==null){
            return null;
        }
        ArrayList<StoragePushMessage> list = new ArrayList<>(total);
        queue.drainTo(list, total);
        return list;
    }

    @Override
    public void persistence(PushMessage msg) {
        try (MongoClient mongoClient = MongoClients.create(Constants.connection)) {
            MongoDatabase database = mongoClient.getDatabase(Constants.database);
            MongoCollection<Document> ordinary = database.getCollection("ordinary");
            Method[] methods = PushMessage.class.getMethods();
            Document document = new Document();
            for (Method method : methods) {
                String methodName = method.getName();
                if (methodName.startsWith("get") && !methodName.equals("getClass")) {
                    String fieldName = getPropertyName(methodName);
                    Object object = method.invoke( msg, null);
                    document.append(fieldName, object);
                }
            }
            ordinary.insertOne(document);
        } catch (InvocationTargetException e) {
            throw new RuntimeException(e);
        } catch (IllegalAccessException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 首字母转小写
     * @param str
     * @return
     */
    private String getPropertyName(String str) {
        String substring = str.substring(3);
        if (Character.isLowerCase(substring.charAt(0))){
            return substring;
        } else{
            return (new StringBuilder()).append(Character.toLowerCase(substring.charAt(0))).append(substring.substring(1)).toString();
        }
    }

    /**
     * 因为设置队列满后抛异常,所以消息肯定都在队列里,此处不需要实现
     * @return
     */
    @Override
    public List<PushMessage> getByDisk() {
        return null;
    }
}
